package org.qy.star_server.core;

import com.google.protobuf.ByteString;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.qy.star_common.message.Message;
import org.qy.star_grpc.core_api.BrokerCoreApiProto;

import java.util.List;
import java.util.Objects;
import java.util.concurrent.*;

/**
 * @author yinger
 * @description TODO
 * @date 2024/3/26 07:40
 **/
@Data
@Slf4j
public class VirtualHost {
    
    @Data
    public static class VirtualHostGloballyUniqueIdentity {
        private String username;
        private String virtualHostName;

        public VirtualHostGloballyUniqueIdentity(String username, String virtualHostName) {
            this.username = username;
            this.virtualHostName = virtualHostName;
        }

        @Override
        public boolean equals(Object object) {
            if (this == object) return true;
            if (object == null || getClass() != object.getClass()) return false;
            VirtualHostGloballyUniqueIdentity that = (VirtualHostGloballyUniqueIdentity) object;
            return Objects.equals(username, that.username) && Objects.equals(virtualHostName, that.virtualHostName);
        }

        @Override
        public int hashCode() {
            return Objects.hash(username, virtualHostName);
        }
    }

    private int deleted;                                                            // 虚拟主机是否被删除（db标记）
    private String username;                                                        // 创建虚拟主机的用户
    private String virtualHostName;                                                 // 虚拟主机名称
    private BlockingQueue<Token> tokenQueue = new LinkedBlockingQueue<>();          // 令牌队列
    private ExecutorService workPool = Executors.newFixedThreadPool(10);   // 线程池 执行回调
    private Thread scanThread = null;                                                // 扫描线程
    private boolean scanThreadWorking = false;                                       // 扫描线程是否工作

    public VirtualHost(String username, String virtualHostName) {
        this.virtualHostName = virtualHostName;
        this.username = username;
    }

    public VirtualHost() {
    }


    @Override
    public String toString() {
        return "VirtualHost{" +
                ", virtualHostName='" + virtualHostName + '\'' +
                ", username='" + username + '\'' +
                '}';
    }

    @Override
    public boolean equals(Object object) {
        if (this == object) return true;
        if (object == null || getClass() != object.getClass()) return false;
        VirtualHost that = (VirtualHost) object;
        return Objects.equals(username, that.username) && Objects.equals(virtualHostName, that.virtualHostName);
    }

    @Override
    public int hashCode() {
        return Objects.hash(username, virtualHostName);
    }

    public void startScanThread() {
        if (!scanThreadWorking) {
            scanThread = new Thread(() -> {
                System.out.println("scan thread start");
                while (true) {
                    try {
                        // 1. 如果令牌队列为空，则等待
                        Token token = tokenQueue.take();
                        Queue queue =  BrokerServer.memory.selectQueue(token.getUsername(), token.getVirtualHostName(), token.getQueueName());
                        if (queue.getEnvList().isEmpty()) {
                            // 该队列没有消费者，将令牌放回队列
                            tokenQueue.put(token);
                            continue;
                        }
                        // 该队列有消费者，进行轮询，选择一个消费者消费消息
                        ConsumerEnv consumerEnv = queue.choseConsumerEnv();
                        // 推送消息到消费者
                        Message message =  BrokerServer.memory.pollMessage(token.getUsername(), token.getVirtualHostName(), token.getQueueName());
                         BrokerServer.memory.insertMessageToQueue(token.getUsername(), token.getVirtualHostName(), token.getQueueName(), message);
                        sendMessage(consumerEnv, message);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            });
            scanThread.start();
            scanThreadWorking = true;
        }
    }

    public void sendMessage(ConsumerEnv consumerEnv, Message message) {
        try {
            BrokerCoreApiProto.Message msg = BrokerCoreApiProto.Message.newBuilder()
                    .setDeliveryMode(message.getDeliveryMode())
                    .setMessageId(message.getMessageId())
                    .setPayload(ByteString.copyFrom(message.getBody()))
                    .setRoutingKey(message.getRoutingKey()).build();
             BrokerServer.memory.insertMessageToWaitAckMessageQueue(consumerEnv.getUsername(), consumerEnv.getVirtualHostName(), consumerEnv.getQueueName(), message);
            consumerEnv.getResponseStreamChannel().onNext(msg);
            if (consumerEnv.isAutoAck()) {
                // 删除待确认消息
                 BrokerServer.memory.deleteMessageFromWaitAckMessageQueue(consumerEnv.getUsername(), consumerEnv.getVirtualHostName(), consumerEnv.getQueueName(), message);
                // 删除磁盘数据
                 BrokerServer.disk.deleteMessage(consumerEnv.getUsername(), consumerEnv.getVirtualHostName(), consumerEnv.getQueueName(), message);
                // 删除消息队列数据
                 BrokerServer.memory.deleteMessageFromQueue(consumerEnv.getUsername(), consumerEnv.getVirtualHostName(), consumerEnv.getQueueName(), message);
                // 删除消息中心数据
                 BrokerServer.memory.deleteMessageFromCenter(message);
            }
        } catch (Exception e) {
            log.error("send message error ... ");
        }
    }
}
